package com.intct.hbase.dml;

import com.intct.util.ConnectionHBaseUtils;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.CompareOperator;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * @author gufg
 * @since 2025/6/13
 */
public class FliterScanData {
    public static void main(String[] args) throws IOException {
        // 连接HBase
        Connection connection = ConnectionHBaseUtils.getConnection();

        // 获取Admin实例
        Admin admin = connection.getAdmin();

        // TabelName实例
        TableName tableName = TableName.valueOf("intct_80_api", "t4");

        // 获取Table实例
        Table table = connection.getTable(tableName);

        try {
            // 判断是否存在
            if (admin.tableExists(tableName)) {
                Scan scan = new Scan();
//                // 查询范围  == 开始Rowkey
//                scan.withStartRow(Bytes.toBytes("1003"));
//
//                // 查询范围  == 结束Rowkey  注意：第二个参数默为false, true时是包含1006一行数据，如果false不包含1006一行数据
//                scan.withStopRow(Bytes.toBytes("1009"), true);
//
//                // 查询数据集中TopN
//                scan.setLimit(3);

                // 创建过滤列表
                FilterList filterList = new FilterList();

                // 单值过滤条件
                SingleColumnValueFilter singleColumnValueFilter =
                        new SingleColumnValueFilter(
                                Bytes.toBytes("f1"), Bytes.toBytes("age"), CompareOperator.EQUAL, Bytes.toBytes("28")
                        );

                RegexStringComparator regexStringComparator = new RegexStringComparator("100.*");

                RowFilter rowFilter = new RowFilter(CompareOperator.EQUAL, regexStringComparator);

                List<Filter> fList = Arrays.asList(singleColumnValueFilter, rowFilter);

                /**
                 * 常用过滤器：
                 *      FamilyFilter    列族过滤
                 *
                 *      QualiferFilter  列限定符过滤
                 *
                 *      RowFilter       RowKey过滤
                 *
                 *      SingleColumnValueFilter 单值过滤
                 *
                 *      ColumnValueFilter   值过滤
                 *
                 *      PageFilter          分页过滤
                 */

                // 将单值过滤增加到过滤列表
                filterList.addFilter(fList);

                // 所有过滤条件增加到scan实例
                scan.setFilter(filterList);



                // 查询数据
                ResultScanner scanner = table.getScanner(scan);

                for (Result result : scanner) {
                    Cell[] cells = result.rawCells();
                    for (Cell cell : cells) {
                        String rowKey = Bytes.toString(CellUtil.cloneRow(cell));
                        String family = Bytes.toString(CellUtil.cloneFamily(cell));
                        String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                        String value = Bytes.toString(CellUtil.cloneValue(cell));
                        System.out.println("row : " + rowKey + "; family : " + family + "; qualifier : " + qualifier + "; value : " + value);
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 关闭资源
            if (admin != null) {
                admin.close();
            }
            if (table != null) {
                table.close();
            }
        }

    }
}
